package com.doit.spark.day03

import com.doit.spark.day01.utils.SparkUtil
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

/**
 * @DATE 2022/1/5/10:51
 * @Author MDK
 * @Version 2021.2.2
 * */
object C04_GroupByKey {
  def main(args: Array[String]): Unit = {
    val sc: SparkContext = SparkUtil.getSc
    val rdd: RDD[String] = sc.textFile("data/word.txt")
    val rdd2: RDD[String] = rdd.flatMap(_.split("\\s+"))
    //打印分区中的所有数据 数据划分依据切片行范围进行
//    rdd2.glom().map(_.toList).foreach(println)

    //将数据变成k,v形式
    val wordone = rdd2.map((_, 1))
    val group: RDD[(String, Iterable[Int])] = wordone.groupByKey(3)
    //打印组装后的数据
    wordone.glom().map(_.toList).foreach(println)
    println(group.getNumPartitions)

    println("-----------------打印重新分区后的结果-------------------------")
    //分区数改为3个后  底层重新进行分区  产生shuffle  默认调用HashPartitioner的分区器
    group.map(_._1).glom().map(_.toList).foreach(println)

    println("---------------------统计单词出现的次数-----------------------")
    group.map(tp=>{
      var cnt =0
      val word = tp._1
      for (elem <- tp._2) {
        cnt+=1
      }
      (word,cnt)
    }).foreach(println)
  }
}
